from pyspark import SparkConf, SparkContext
import os

if __name__ == '__main__':
    os.environ["PYSPARK_PYTHON"] = "D:/Python/Python310/python.exe"
    os.environ["HADOOP_HOME"] = "D:/Python/hadoop-3.0.0"

    conf = SparkConf().setMaster("local[*]").setAppName("create rdd")
    sc = SparkContext(conf=conf)
    rdd = sc.textFile("D:/BaiduNetdiskDownload/heima/Python/第15章资料/资料/search_log.txt")
    # 打印输出：热门搜索时间段（小时精度）Top3
    rdd_map = rdd.map(lambda x: (x.split("\t")[0][:2], 1)) \
        .reduceByKey(lambda a, b: a + b) \
        .sortBy(lambda x: x[1], ascending=False, numPartitions=1).take(3)
    print("需求1的结果：", rdd_map)
    # 打印输出：热门搜索词Top3
    rdd_map2 = rdd.map(lambda x: (x.split("\t")[2], 1)) \
        .reduceByKey(lambda a, b: a + b) \
        .sortBy(lambda a: a[1], ascending=False, numPartitions=1) \
        .take(3)
    print("需求2的结果：", rdd_map2)
    # 打印输出：统计黑马程序员关键字在哪个时段被搜索最多
    rdd_map3 = rdd.map(lambda a: a.split("\t")).filter(lambda a: a[2] == '黑马程序员').map(
        lambda a: (a[0][:2], 1)).reduceByKey(
        lambda a, b: a + b).sortBy(
        lambda a: a[0], ascending=False, numPartitions=1).take(1)
    print("需求3的结果：", rdd_map3)
    # 将数据转换为JSON格式，写出为文件
    rdd.map(lambda a: a.split("\t")).map(
        lambda a: {"time": a[0], "user_id": a[1], "key_word": a[2], "rank1": a[3], "rank2": a[4],
                   "url": a[5]}).saveAsTextFile("D:/outfile1")

